package com.zhengb.flink.study.datastream.transformation;

import com.zhengb.flink.study.datastream.datasource.RandomParallelSource;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;

/**
 * 自定义非并行 source function
 * <p>
 * DataStream -> DataStream
 *
 * @author zhengbo
 * @date 2019/12/19
 */
@Slf4j
public class FlatMapTransformation {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.addSource(new RandomParallelSource()).setParallelism(10).flatMap(new FlatMapFunction<Long, Long>() {

            @Override
            public void flatMap(Long value, Collector<Long> out) throws Exception {
                log.info("receive data:{}", value);
                out.collect(value * 3);
            }
        }).setParallelism(5).addSink(new RichSinkFunction<Long>() {
            @Override
            public void invoke(Long value, Context context) throws Exception {
                log.info("sink data:{}", value);
            }
        });

        env.execute("FlatMapTransformation");
    }


}
